package com.chinamobile.iot.handler.redis;

import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.utils.NameThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by szl on 2017/7/18.
 */
@Component
@ConditionalOnExpression("${mqtt.distribute}")
public class RedisService implements ApplicationContextAware {

    private Logger logger = LoggerFactory.getLogger(RedisService.class);

    @Autowired
    private RedisListener redisListener;
    @Autowired
    private MqttConfiguration mqttConfiguration;

    /**
     * the message listener container
     */
    private RedisMessageListenerContainer container;
    /**
     * the thread pool of redis worker
     */
    private ExecutorService redisWorkerExecutor;
    /**
     * the redis subscribe listener
     */
    private MessageListenerAdapter messageListenerAdapter;

    private ApplicationContext context;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }

    @PostConstruct
    public void init() {
        if (mqttConfiguration.isDistribute()) {
            RedisConnectionFactory factory = context.getBean(RedisConnectionFactory.class);
            container = context.getBean(RedisMessageListenerContainer.class);
            container.setConnectionFactory(factory);

            messageListenerAdapter = new MessageListenerAdapter();
            messageListenerAdapter.setDelegate(redisListener);
            messageListenerAdapter.setDefaultListenerMethod("receiveMessage");
            messageListenerAdapter.afterPropertiesSet();


            redisWorkerExecutor = Executors.newFixedThreadPool(2, new NameThreadFactory("redis-sub-worker-"));
            container.setTaskExecutor(redisWorkerExecutor);
            container.setRecoveryInterval(1000);
            container.afterPropertiesSet();
        }
    }

    @PreDestroy
    public void destroy() {
        redisWorkerExecutor.shutdown();
    }

    public void addListener(String topic) {
        container.addMessageListener(messageListenerAdapter, new PatternTopic(topic));
        logger.info("add redis listener on topic {}", topic);
    }

    public void removeListen(String topic) {
        container.removeMessageListener(messageListenerAdapter, new PatternTopic(topic));
        logger.info("remove redis listener on topic {}", topic);
    }
}
